package com.etl.component.basic;

import cn.hutool.core.util.StrUtil;
import com.google.common.collect.Lists;
import com.component.api.constants.ComponentCodeConstants;
import com.component.api.constants.ComponentConstants;
import com.component.api.excepiton.ComponentException;
import com.component.api.model.PortData;
import com.component.api.model.ProcessResult;
import com.component.api.model.data.ColumnData;
import com.component.api.model.data.SetData;
import com.component.api.model.dto.DefaultColumnDTO;
import com.component.api.model.param.ParamPair;
import com.component.api.model.param.TableColumn;
import com.component.api.utils.TypesUtils;
import com.common.utils.tool.StringUtil;
import com.etl.component.common.AbstractFlinkComponent;
import org.apache.flink.api.common.typeinfo.TypeInformation;
import org.apache.flink.api.java.DataSet;
import org.apache.flink.api.java.ExecutionEnvironment;
import org.apache.flink.api.java.Utils;
import org.apache.flink.api.java.operators.MapOperator;
import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.api.java.typeutils.RowTypeInfo;
import org.apache.flink.types.Row;

import java.util.ArrayList;
import java.util.Comparator;
import java.util.List;
import java.util.Map;
import java.util.stream.Collectors;

/**
 * 描述：
 * 缺省值填充
 * @author xianggj
 * @Date 2021/12/3 18:42
 **/
public class DefaultValuePaddingComponent  extends AbstractFlinkComponent {
    @Override
    public ProcessResult process(List<PortData> datas, List<ParamPair> paramPairs) {
        Map<String, Object> paramMap = paramToObject(paramPairs);
        List<DefaultColumnDTO> defaultColumnDTOS =
                (List<DefaultColumnDTO>) paramMap.get(ComponentConstants.COLUMNS);
        SetData input = (SetData)datas.get(0).getValue();
        DataSet<Row> dataset = (DataSet<Row>) input.getDataset();
        List<TableColumn> columns = input.getColumns();
        setIndexAndRelyIndex(defaultColumnDTOS, columns);
        RowTypeInfo type = (RowTypeInfo)dataset.getType();
        TypeInformation<?>[] fieldTypes = type.getFieldTypes();

        //字段值计算
        List<Integer> interList = autoRuleVal(defaultColumnDTOS, dataset, fieldTypes);

        DataSet<Row> result = new MapOperator<Row, Row>(dataset, type, e->{
            for (DefaultColumnDTO defaultColumnDTO : defaultColumnDTOS) {
                int index = defaultColumnDTO.getIndex();
                TypeInformation<?> fieldType = (TypeInformation<?>) fieldTypes[index];
                Object field = e.getField(index);
                Object resVal = null;
                //数值计算迁移到统一输出了
                if (defaultColumnDTO.getRule().equalsIgnoreCase(DefaultColumnDTO.AUTO)) {
                    if (DefaultColumnDTO.INTER.equalsIgnoreCase(defaultColumnDTO.getRuleVal())){
                        continue;
                    }
                    if (field != null){
                        continue;
                    }
                    //取最大值 最小值 中值 平均值 都可以在这里 但是插值不行
                    resVal = defaultColumnDTO.getRuleFormula();

                } else if (defaultColumnDTO.getRule().equalsIgnoreCase(DefaultColumnDTO.RELY)) {
                    if (field != null){
                        continue;
                    }
                    DefaultColumnDTO.Rely relyVal = defaultColumnDTO.getRelyVal();
                    int relyIndex = relyVal.getIndex();
                    Object relyField = e.getField(relyIndex);
                    if (relyField == null){
                        continue;
                    }
                    double val = Double.parseDouble(relyField.toString());
                    resVal = relyVal.getFunction().calc(val);
                } else {
                    resVal = field;
                }
                Object covert = null;
                //异常后默认为空
                try {
                    covert = covert(resVal, fieldType);
                } catch (Exception ex) {
                    ex.printStackTrace();
                }
                e.setField(index, covert);
            }
            return e;
        }, Utils.getCallLocationName());
        if (interList.size() > 0){
            try {
                //这会断流 哎但是没办法
                final ExecutionEnvironment env = result.getExecutionEnvironment();
                List<Row> collect = result.collect();
                //当只有1条和0条数据时 不需要计算
                if (collect.size() <= 1){
                    result = env.fromCollection(collect, type);
                } else{
                    InterFunction interFunction =
                            new InterFunction(interList.toArray(new Integer[0]), fieldTypes, collect.size());
                    //插值最后计算
                    for (int i = 0; i < collect.size()-1 ; i++) {
                        Row row1 = collect.get(i);
                        Row row2 = collect.get(i+1);
                        interFunction.reduce(new Tuple2<>(i,row1), new Tuple2<>(i+1,row2));
                    }
                    result = env.fromCollection(interFunction.getResList(), type);
                }
            } catch (Exception e) {
                e.printStackTrace();
                throw new ComponentException("插值计算错误", e);
            }
        }

        return execute(result, columns, "执行缺省值填充操作");
    }


    /**
     * 自动添值计算 插值计算这里做不了
     * @param defaultColumnDTOS
     * @param dataset
     * @param fieldTypes
     * @return 返回插值计算列 如需要插值计算的列为 第1列和第3列
     */
    private List<Integer> autoRuleVal(List<DefaultColumnDTO> defaultColumnDTOS,
                                      DataSet<Row> dataset, TypeInformation<?>[] fieldTypes) {
        List<Integer> list = new ArrayList<>();
        //自动填充特殊处理
        try {
            List<Row> collect = dataset.collect();
            for (int i = 0; i < defaultColumnDTOS.size(); i++) {
                DefaultColumnDTO defaultColumnDTO = defaultColumnDTOS.get(i);
                //只有自动填充走这里
                if (!defaultColumnDTO.getRule().equalsIgnoreCase(DefaultColumnDTO.AUTO)){
                    continue;
                }
                String rule = defaultColumnDTO.getRuleVal();
                if (StrUtil.isBlank(rule)) {
                    continue;
                }
                final int indexI = defaultColumnDTO.getIndex();
                TypeInformation<?> fieldType =  fieldTypes[indexI];
                switch (rule) {
                    // 插值计算 这里做不了 提取出来
                    case DefaultColumnDTO.INTER:
                        list.add(indexI);
                        break;
                    case DefaultColumnDTO.MAX:
                        Object max = getMax(fieldType, collect, indexI);
                        defaultColumnDTO.setRuleFormula(max==null? null : max.toString());
                        break;
                    case DefaultColumnDTO.MIN:
                        Object min = getMin(fieldType, collect, indexI);
                        defaultColumnDTO.setRuleFormula(min==null? null : min.toString());
                        break;
                    case DefaultColumnDTO.AVG:
                        Object avg = getAvg(fieldType, collect, indexI);
                        defaultColumnDTO.setRuleFormula(avg==null? null : avg.toString());
                        break;
                    case DefaultColumnDTO.MID:
                        Object mid = getMid(fieldType, collect, indexI);
                        defaultColumnDTO.setRuleFormula(mid==null? null : mid.toString());
                        break;
                    default:
                        continue;
                }
            }
        }catch (Exception e) {
            e.printStackTrace();
        }
        return list;
    }


    /**
     * 返回平均值
     * @param basicTypeInfo
     * @return
     */
    public Object getAvg(TypeInformation basicTypeInfo, List<Row> collect, final int indexI) {
        switch (TypesUtils.unifiedType(typeToInt(basicTypeInfo))) {
            case INTEGER:
                return  collect.stream().map(e -> e.getField(indexI))
                        .filter(e -> e != null)
                        .collect(Collectors.averagingInt(e -> Integer.valueOf(e.toString())));
            case LONG:
                return  collect.stream().map(e -> e.getField(indexI))
                        .filter(e -> e != null)
                        .collect(Collectors.averagingLong(e ->  Long.valueOf(e.toString())));
            case DOUBLE:
                return  collect.stream().map(e -> e.getField(indexI))
                        .filter(e -> e != null)
                        .collect(Collectors.averagingDouble(e ->  Double.valueOf(e.toString())));
            default:
                return null;
        }
    }

    /**
     * 返回最大值
     * @param basicTypeInfo
     * @return
     */
    public Object getMax(TypeInformation basicTypeInfo, List<Row> collect, final int indexI) {
        switch (TypesUtils.unifiedType(typeToInt(basicTypeInfo))) {
            case INTEGER:
                return collect.stream().map(e -> e.getField(indexI))
                    .filter(e -> e != null)
                    .max(Comparator.comparingInt(a -> Integer.valueOf(a.toString()))).orElseGet(null);
            case LONG:
                return collect.stream().map(e -> e.getField(indexI))
                        .filter(e -> e != null)
                        .max(Comparator.comparingLong(a -> Long.valueOf(a.toString()))).orElseGet(null);
            case DOUBLE:
                return collect.stream().map(e -> e.getField(indexI))
                        .filter(e -> e != null)
                        .max(Comparator.comparingDouble(a -> Double.valueOf(a.toString()))).orElseGet(null);
            default:
                return null;
        }
    }


    /**
     * 返回最小值
     * @param basicTypeInfo
     * @return
     */
    public Object getMin(TypeInformation basicTypeInfo, List<Row> collect, final int indexI) {
        switch (TypesUtils.unifiedType(typeToInt(basicTypeInfo))) {
            case INTEGER:
                return collect.stream().map(e -> e.getField(indexI))
                        .filter(e -> e != null)
                        .min(Comparator.comparingInt(a -> Integer.valueOf(a.toString()))).orElseGet(null);
            case LONG:
                return collect.stream().map(e -> e.getField(indexI))
                        .filter(e -> e != null)
                        .min(Comparator.comparingLong(a -> Long.valueOf(a.toString()))).orElseGet(null);
            case DOUBLE:
                return collect.stream().map(e -> e.getField(indexI))
                        .filter(e -> e != null)
                        .min(Comparator.comparingDouble(a -> Double.valueOf(a.toString()))).orElseGet(null);
            default:
                return null;
        }
    }


    /**
     * 返回中值
     * @param basicTypeInfo
     * @return
     */
    public Object getMid(TypeInformation basicTypeInfo, List<Row> collect, final int indexI) {
        switch (TypesUtils.unifiedType(typeToInt(basicTypeInfo))) {
            case INTEGER:
                List<Integer> intSorted = collect.stream().map(e -> e.getField(indexI))
                        .filter(e -> e != null)
                        .map(e->Integer.valueOf(e.toString()))
                        .sorted(Comparator.comparingInt(a -> (int) a)).collect(Collectors.toList());
                int isize = intSorted.size();
                if (isize == 0){
                    return null;
                }
                return isize%2 == 1? intSorted.get(isize/2): Math.round((intSorted.get(isize/2-1) + intSorted.get(isize/2))/2.0) ;
            case LONG:
                List<Long> lonSorted = collect.stream().map(e -> e.getField(indexI))
                        .filter(e -> e != null)
                        .map(e->Long.valueOf(e.toString()))
                        .sorted(Comparator.comparingLong(a -> (long) a)).collect(Collectors.toList());
                int lSize = lonSorted.size();
                if (lSize == 0){
                    return null;
                }
                return lSize%2 == 1? lonSorted.get(lSize/2): Math.round((lonSorted.get(lSize/2-1) + lonSorted.get(lSize/2))/2.0);
            case DOUBLE:
                List<Double> douSorted = collect.stream().map(e -> e.getField(indexI))
                        .filter(e -> e != null)
                        .map(e-> Double.valueOf(e.toString()))
                        .sorted(Comparator.comparingDouble(a -> (double) a)).collect(Collectors.toList());
                int dsize = douSorted.size();
                if (dsize == 0){
                    return null;
                }
                return dsize%2 == 1? douSorted.get(dsize/2): (douSorted.get(dsize/2-1) + douSorted.get(dsize/2))/2 ;
            default:
                return null;
        }
    }

    /**
     * 返回比较器
     * @param basicTypeInfo
     * @return
     */
    public Comparator<? super Object> getComparator(TypeInformation basicTypeInfo) {
        switch (TypesUtils.unifiedType(typeToInt(basicTypeInfo))) {
            case INTEGER:
                return Comparator.comparingInt(a -> Integer.valueOf(a.toString()));
            case LONG:
                return Comparator.comparingLong(a -> Long.valueOf(a.toString()));
            case DOUBLE:
                return Comparator.comparingDouble(a -> Double.valueOf(a.toString()));
            default:
                return Comparator.comparing(a -> StringUtil.toString(a));
        }
    }

    /**
     * 缺失值转换的特殊处理
     * @param columnDTOS
     * @param columns
     */
    public void setIndexAndRelyIndex(List<DefaultColumnDTO> columnDTOS, List<TableColumn> columns) {
        List<String> colnames = columns.stream().map(e -> e.getColumnName()).collect(Collectors.toList());
        for (DefaultColumnDTO columnDTO : columnDTOS) {
            int i = colnames.indexOf(columnDTO.getField());
            if (i == -1){
                throw new ComponentException("数据来源列和输出列不对应，数据来源列中缺少列" + columnDTO.getField());
            }
            columnDTO.setIndex(i);
            if (DefaultColumnDTO.RELY.equalsIgnoreCase(columnDTO.getRule())
              && columnDTO.getRelyVal() != null){
                DefaultColumnDTO.Rely relyVal = columnDTO.getRelyVal();
                int j = colnames.indexOf(relyVal.getField());
                if (i == -1){
                    throw new ComponentException("数据来源列和依赖列不对应，数据来源列中缺少列" + relyVal.getField());
                }
                relyVal.setIndex(j);
            }
        }
    }

    @Override
    public List<PortData> previewOutputColumns(List<PortData> datas, List<ParamPair> paramPairs) {
        ColumnData input = (ColumnData)datas.get(0).getValue();
        List<TableColumn> columns = input.getColumns();
        return Lists.newArrayList(new PortData(getRule(), new ColumnData(columns)));
    }

    @Override
    public String getRule() {
        return ComponentCodeConstants.DEFAULT_PADDING;
    }
}
